// 模拟Java的future模式
package channel

import (
	"errors"
	"fmt"
	"log"
	"sync"
	"time"
)

//多个FutureListener按添加顺序依次执行
type FutureListener interface {
	// FutureTask结束（包括panic）后回调
	operationComplete(f Future)
}

type FutureListenerOnce struct {
	once sync.Once
	fl   FutureListener
}

func (fo *FutureListenerOnce) operationComplete(f Future) () {
	fo.once.Do(func() {

		defer func() {
			if e := recover(); e != nil {
				log.Fatalf(fmt.Sprintf("FutureTask panic,%v", e))
			}
		}()
		fo.fl.operationComplete(f)
	})
}

func newFutureListenerOnce(fl FutureListener) *FutureListenerOnce {
	return &FutureListenerOnce{
		once: sync.Once{},
		fl:   fl,
	}
}

type Future interface {
	//阻塞等待返回结果
	Get() (interface{}, error)
	// 阻塞等待返回结果,直到超时
	GetTimeout(timeout time.Duration) (interface{}, error)
	// 添加监听器，返回自身.多个FutureListener按添加顺序依次执行
	AddListener(listener FutureListener) Future
	// 添加监听器，返回自身.多个FutureListener按添加顺序依次执行
	AddListeners(listeners ...FutureListener) Future
	// 移除监听器，返回自身
	RemoveListener(listener FutureListener) Future
	// 添加监听器，返回自身
	RemoveListeners(listeners ...FutureListener) Future
}

const (
	STATE_INIT     = 0
	STATE_RUNNING  = 1
	STATE_COMPLETE = 2
)

type DefaultFuture struct {
	state        int8
	lock         sync.Mutex
	listeners    []FutureListener
	getResult    func(timeout time.Duration) (interface{}, error)
	futureResult *FutureResult
}

func newDefaultFuture(getResult func(timeout time.Duration) (interface{}, error)) *DefaultFuture {
	return &DefaultFuture{
		state:     STATE_INIT,
		lock:      sync.Mutex{},
		getResult: getResult,
		listeners: make([]FutureListener, 0,5),
	}
}

func (f *DefaultFuture) Get() (interface{}, error) {
	return f.GetTimeout(0)
}

func (f *DefaultFuture) GetTimeout(timeout time.Duration) (interface{}, error) {

	if nil != f.futureResult{
		return f.futureResult.result,f.futureResult.err
	}


	r,err:= f.getResult(timeout)
	f.futureResult = &FutureResult{
		result: r,
		err: err,
	}

	return r,err

}

func (f *DefaultFuture) AddListener(listener FutureListener) Future {
	f.lock.Lock()
	defer f.lock.Unlock()
	fo := newFutureListenerOnce(listener)
	f.listeners = append(f.listeners, fo)

	if f.state == STATE_COMPLETE {
		//  如果futureTask已经执行完了才加入FutureListener,则将FutureListener手动执行一遍
		fo.operationComplete(f)
	}
	return f
}

func (f *DefaultFuture) AddListeners(listeners ...FutureListener) Future {
	f.lock.Lock()
	defer f.lock.Unlock()

	//f.listeners = append(f.listeners, listeners...)

	for _, l := range listeners {
		f.AddListener(l)
	}

	return f
}
func (f *DefaultFuture) RemoveListener(listener FutureListener) Future {
	f.lock.Lock()
	defer f.lock.Unlock()
	for i, l := range f.listeners {
		if l == listener {
			f.listeners = append(f.listeners[:i], f.listeners[i+1:]...)
		}
	}
	return f
}
func (f *DefaultFuture) RemoveListeners(listeners ...FutureListener) Future {
	f.lock.Lock()
	defer f.lock.Unlock()

	for _, l := range listeners {
		f.RemoveListener(l)
	}
	return f
}

//
//
// @Param taskF 任务逻辑函数
// @Param args 传给任务逻辑函数的参数
//
// @Return future
//
func Submit(taskF FutureTask) Future {
	c := make(chan FutureResult, 1)


	getResult := func(timeout time.Duration) (r interface{}, err error) {
		defer func() {
			if e := recover(); e != nil {
				r, err = nil, errors.New(fmt.Sprintf("FutureTask panic,%v", e))
			}
		}()

		if timeout > 0 {
			// select 是按顺序判断执行，故先判断是否超时
			select {
			case <-time.After(timeout):
				r, err = nil, errors.New("future timeout")
			case fr := <-c:
				r, err = fr.result, fr.err
			}
		} else {
			fr := <-c
			r, err = fr.result, fr.err
		}

		return r, err
	}

	future := newDefaultFuture(getResult)

	go func() {
		defer close(c)
		future.lock.Lock()
		future.state = STATE_RUNNING
		future.lock.Unlock()

		result, err := taskF.run()
		future.lock.Lock()
		future.state = STATE_COMPLETE
		c <- FutureResult{
			result: result,
			err:    err,
		}
		// 执行回调
		if nil != future.listeners && len(future.listeners) > 0 {
			go func() {
				for _, l := range future.listeners {
					l.operationComplete(future)
				}
			}()
		}

		future.lock.Unlock()


	}()

	return future
}
